package com.apps.sdses.flink141.source;

import com.apps.sdses.flink141.entity.Event;
import com.apps.sdses.flink141.entity.SimpleEvent;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.streaming.connectors.kafka.KafkaDeserializationSchema;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.nio.charset.StandardCharsets;
import java.util.HashMap;

import static org.apache.flink.api.java.typeutils.TypeExtractor.getForClass;

/**
 * author: codejiwei
 * date: 2023/9/3
 * desc:
 **/
public class KafkaEventDeserializationSchema implements KafkaDeserializationSchema<Event> {
    Logger logger = LoggerFactory.getLogger(KafkaEventDeserializationSchema.class);
    private final SimpleStringSchema simpleStringSchema = new SimpleStringSchema();
    private boolean flag = true;
    @Override
    public boolean isEndOfStream(Event nextElement) {
        return false;
    }

    @Override
    public Event deserialize(ConsumerRecord<byte[], byte[]> record) throws Exception {
        SimpleEvent event = new SimpleEvent();
        HashMap<String, String> headerMap = new HashMap<String, String>() {
            {
                put("topic", record.topic());
                put("partition", String.valueOf(record.partition()));
                put("offset", String.valueOf(record.offset()));
                put("timestamp", String.valueOf(record.timestamp()));
                put("processTimestamp", String.valueOf(System.currentTimeMillis()));
                if (record.key() != null) {
                    put("key", new String(record.key(), StandardCharsets.UTF_8));
                } else {
                    put("key", "__null_key__");
                }
            }
        };

        record.headers().forEach(header -> headerMap.put(header.key(), simpleStringSchema.deserialize(header.value())));
        event.setHeaders(headerMap);
        event.setBody(record.value());
        if (flag) {
            logger.info("the first event meta info: " + event.toString());
        }
        flag = false;

        return event;
    }

    @Override
    public TypeInformation<Event> getProducedType() {
        return getForClass(Event.class);
    }
}
